package com.mjf.spark.day09

import java.util.Properties

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.sql.{DataFrame, Dataset, SparkSession}

/**
 * 通过JDBC对MySQL进行读写操作
 */
object SparkSQL01_MySQL {
  def main(args: Array[String]): Unit = {

    // 创建SparkConf配置文件对象
    val conf: SparkConf = new SparkConf().setAppName("SparkSQL").setMaster("local[*]")

    // 创建SparkSQL执行的入口对象  SparkSession
    val spark: SparkSession = SparkSession.builder().config(conf).getOrCreate()

    import spark.implicits._

    /*
    // 从MySQL数据库中读取数据 方式1
    val df: DataFrame = spark.read.format("jdbc")
      .option("url", "jdbc:mysql://hadoop103:3306/test")
      .option("driver", "com.mysql.jdbc.Driver")
      .option("user", "root")
      .option("password", "123456")
      .option("dbtable", "user")
      .load()

    df.show()
    */

    /*
    // 从MySQL数据库中读取数据 方式2
      val df: DataFrame = spark.read.format("jdbc")
      .options(
        Map(
          "url" -> "jdbc:mysql://hadoop103:3306/test?user=root&password=123456",
          "driver" -> "com.mysql.jdbc.Driver",
          "dbtable" -> "user"
        )
      ).load()

    df.show()
    */

    /*
    // 从MySQL数据库中读取数据 方式3
    val props: Properties = new Properties()    // Hashtable的子类
    props.setProperty("user", "root")
    props.setProperty("password", "123456")
    props.setProperty("driver", "com.mysql.jdbc.Driver")
    val df: DataFrame = spark.read.jdbc("jdbc:mysql://hadoop103:3306/test?user=root&password=123456", "user", props)
    df.show()
    */

    // 向MySQL数据库中写入数据
    val sc: SparkContext = spark.sparkContext
    val rdd: RDD[User] = sc.makeRDD(List(User("marry", 19), User("linda", 18)))

    // 将RDD转换为DataFrame
//    val df: DataFrame = rdd.toDF()
//    df.write.format("jdbc").save()

    val ds: Dataset[User] = rdd.toDS()
    ds.write.format("jdbc")
      .option("url", "jdbc:mysql://hadoop103:3306/test")
      .option("driver", "com.mysql.jdbc.Driver")
      .option("user", "root")
      .option("password", "123456")
      .option("dbtable", "user")
      .mode("append")
      .save()

    // 释放资源
    spark.close()

  }
}

case class User(name:String, age:Int)